package org.jetlinks.pro.device.service.task;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.hswebframework.ezorm.core.dsl.Query;
import org.hswebframework.web.api.crud.entity.PagerResult;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.hswebframework.web.id.IDGenerator;
import org.jetlinks.core.cluster.ClusterManager;
import org.jetlinks.core.device.DeviceRegistry;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.Headers;
import org.jetlinks.pro.PropertyConstants;
import org.jetlinks.pro.device.entity.DeviceInstanceEntity;
import org.jetlinks.pro.device.enums.TaskState;
import org.jetlinks.pro.device.service.LocalDeviceInstanceService;
import org.jetlinks.pro.gateway.DeviceMessageUtils;
import org.reactivestreams.Subscription;
import org.springframework.stereotype.Service;
import reactor.core.Disposable;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.core.scheduler.Schedulers;

import javax.annotation.Nonnull;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.function.Function;
import java.util.function.Supplier;

@Service
@AllArgsConstructor
@Slf4j
public class DeviceMessageTaskService {
    private final DeviceMessageTaskRepository taskRepository;

    private final LocalDeviceInstanceService deviceInstanceService;

    private final DeviceRegistry registry;

    private final ClusterManager clusterManager;

    private final TaskSubscriber subscriber = new TaskSubscriber(this::executeTask);

    /**
     * 分页查询任务信息
     *
     * @param query 动态查询条件
     * @return 分页查询结果
     */
    public Mono<PagerResult<DeviceMessageTask>> queryPager(QueryParamEntity query) {
        return taskRepository.queryPager(query);
    }

    /**
     * 动态查询任务信息
     *
     * @param query 查询条件
     * @return 查询结果
     */
    public Flux<DeviceMessageTask> query(QueryParamEntity query) {
        return taskRepository.query(query);
    }

    /**
     * 更新指定Task的状态
     *
     * @param id    taskId
     * @param state 状态
     * @return void
     * @see TaskState
     */
    public Mono<Void> changeSate(List<String> id, TaskState state) {
        return taskRepository
            .changeState(id, state)
            .doOnSuccess(ignore -> subscriber.tryRequest());
    }

    /**
     * 根据taskId获取task信息
     *
     * @param taskId taskId
     * @return Task信息
     * @see DeviceMessageTask
     */
    public Mono<DeviceMessageTask> findTask(String taskId) {
        return taskRepository
            .findById(taskId);
    }

    /**
     * 立即执行task（发送任务对应的指令给设备）,如果执行失败,错误会传递到下游.
     *
     * @param taskId taskId
     * @return void
     * @see org.jetlinks.core.exception.DeviceOperationException
     */
    public Mono<Void> executeTask(String taskId) {
        return taskRepository
            .findById(taskId)
            .flatMap(this::executeTask);
    }

    /**
     * 使用任务信息执行任务
     *
     * @param task 任务
     * @return void
     */
    public Mono<Void> executeTask(DeviceMessageTask task) {
        return registry
            .getDevice(task.getDeviceId())
            .flatMap(device -> device
                .messageSender()
                .send(task.downstreamToMessage())
                .then());
    }

    @SuppressWarnings("all")
//    @Transactional
    public Mono<Integer> createTask(QueryParamEntity deviceQuery,
                                    Supplier<DeviceMessage> templateSupplier) {

        return deviceInstanceService
            .createQuery()
            .setParam(deviceQuery)
            .select(DeviceInstanceEntity::getId, DeviceInstanceEntity::getName, DeviceInstanceEntity::getProductId)
            .fetch()
            .map(deviceInstance -> {
                DeviceMessage template = templateSupplier.get();
                DeviceMessageUtils.trySetDeviceId(template, deviceInstance.getId());
                DeviceMessageUtils.trySetMessageId(template, IDGenerator.SNOW_FLAKE_STRING::generate);

                template.addHeader(Headers.productId, deviceInstance.getProductId());
                template.addHeader(PropertyConstants.deviceName, deviceInstance.getName());
                return DeviceMessageTask.newTask(template);
            })
            .buffer(200)
            .flatMap(list -> this.createTask(list).thenReturn(list.size()))
            .reduce(Math::addExact);
    }

    public Mono<Void> createTask(List<DeviceMessageTask> tasks) {
        for (DeviceMessageTask task : tasks) {
            task.setState(TaskState.wait);
            task.setServerId(clusterManager.getCurrentServerId());
        }
        return taskRepository
            .addTasks(tasks);
    }

    private Flux<DeviceMessageTask> findWaitSendTask(long total) {
        return QueryParamEntity
            .newQuery()
            .when(total >= Integer.MAX_VALUE, Query::noPaging)
            .when(total < Integer.MAX_VALUE, query -> query.doPaging(0, (int) total))
            .where(DeviceMessageTask::getServerId, clusterManager.getCurrentServerId())
            .in(DeviceMessageTask::getState, TaskState.wait)
            .execute(taskRepository::query);
    }

    @PostConstruct
    public void init() {
        Flux
            .<DeviceMessageTask>create(sink -> {
                AtomicBoolean requesting = new AtomicBoolean();

                sink.onRequest(total -> {
                    try {
                        //正在发送指令就忽略掉
                        if (requesting.compareAndSet(false, true)) {
                            //查找等待发送的指令
                            findWaitSendTask(total)
                                .doOnNext(sink::next)
                                .doFinally(s -> requesting.set(false))
                                .subscribe();
                        }
                    } catch (Throwable e) {
                        requesting.set(false);
                        log.error("request device message task error", e);
                    }
                });
            })
            .subscribe(subscriber);

    }

    @PreDestroy
    public void shutdown(){
        subscriber.dispose();
    }

    static class TaskSubscriber extends BaseSubscriber<DeviceMessageTask> {
        static final AtomicIntegerFieldUpdater<TaskSubscriber> REMAINDER
            = AtomicIntegerFieldUpdater.newUpdater(TaskSubscriber.class, "remainder");
        //每次请求任务的数量
        private static final int batch = Integer.getInteger("device.message.task.bach", 20);

        private volatile int remainder;

        private final Function<DeviceMessageTask, Mono<Void>> handler;

        private Disposable disposable;

        public TaskSubscriber(Function<DeviceMessageTask, Mono<Void>> handler) {
            this.handler = handler;
        }

        @Override
        protected void hookOnSubscribe(@Nonnull Subscription subscription) {
//            request();
            //定时去请求
            disposable = Schedulers
                .parallel()
                .schedulePeriodically(() -> {
                    if (remainder <= 0) {
                        request();
                    }
                }, 20, 10, TimeUnit.SECONDS);
        }

        private void request() {
            //从上游请求任务
            upstream().request(batch);
        }

        @Override
        protected void hookFinally(@Nonnull SignalType type) {
            disposable.dispose();
        }

        private void tryRequest() {
            if (REMAINDER.decrementAndGet(this) <= 0) {
                request();
            }
        }

        @Override
        protected void hookOnNext(@Nonnull DeviceMessageTask value) {
            try {
                REMAINDER.incrementAndGet(this);
                handler.apply(value)
                       .onErrorResume(err -> Mono.empty())
                       .doFinally(signalType -> tryRequest())
                       .subscribe();
            } catch (Throwable err) {
                log.error(err.getMessage(), err);
                tryRequest();
            }
        }
    }
}
